package com.supreme.simpleserver.controller;

import com.supreme.commonboot.utils.AjaxResp;
import com.supreme.simpleserver.consumer.entity.KafkaMsg;
import org.jetbrains.annotations.NotNull;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;

@RestController
@RequestMapping(value = "producer")
public class ProducerController {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @PostMapping(value = "kafka-producer/{topic}")
    public AjaxResp kafkaProducer(@PathVariable(value = "topic") String topic,
                                  @RequestBody KafkaMsg bodyData) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, bodyData.toString());
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(@NotNull Throwable ex) {
                System.out.println("onFailure = " + ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println("onSuccess = " + result);
            }
        });

        return AjaxResp.instance();
    }
}
